package top.lzzly.sync.kafka.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import top.lzzly.sync.kafka.config.Config;
import top.lzzly.sync.kafka.server.IEsService;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * @Author : smq
 * @Description: 消费者
 * @Date : 2018/9/19  09:48
 * @Modified By :
 */
@Slf4j
@Component
public class JsonConsumer {

    @Value("${es.data.format.chargeChannel}")
    String chargeChannelFormat;
    @Value("${es.data.format.chargeDevice}")
    String chargeDeviceFormat;
    @Value("${es.data.format.chargeOrder}")
    String chargeOrderFormat;

    @Autowired
    private IEsService esServiceImpl;

    @KafkaListener(topics = Config.KAFKA_JSON_TOPICS, id = Config.KAFKA_JSON_ID, containerFactory = "batchFactory")
    public void listen(List<ConsumerRecord<?, ?>> list) throws IOException {
        List<String> messages = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : list) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            // 获取消息
            kafkaMessage.ifPresent(o -> messages.add(o.toString()));
        }
        if (messages.size() > 0) {
            log.info("获取消息 = " + messages);
            // 更新索引
            updateES(messages);
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取消息JSON解析格式
     *
     * @param tableName
     * @return
     */
    private String getJsonFormat(String tableName) {
        String format = "";
        switch (tableName) {
            case "charge_device": {
                format = chargeDeviceFormat;
                break;
            }
            case "charge_channel": {
                format = chargeChannelFormat;
                break;
            }
            case "charge_order": {
                format = chargeOrderFormat;
                break;
            }
        }
        return format;
    }


    /**
     * 获取解析后的ES对象
     *
     * @param message
     * @param tableName
     * @return
     */
    private JSONObject getESObject(JSONObject message, String tableName) {
        JSONObject resultObject = new JSONObject();
        String format = getJsonFormat(tableName);
        if (!format.isEmpty()) {
            JSONObject jsonFormatObject = JSON.parseObject(format);
            for (String key : jsonFormatObject.keySet()) {
                resultObject.put(key, message.get(key));
            }
        }
        return resultObject;
    }


    /**
     * 更新ES索引
     *
     * @param messages
     */
    private void updateES(List<String> messages) throws IOException {
        for (String message : messages) {
            JSONObject result = null;
            try {
                result = JSON.parseObject(message);
            } catch (Exception e) {
                continue;
            }
            // 获取事件类型 event:"wtv3.videos.insert"
            String event = (String) result.get("event");
            String[] eventArray = event.split("\\.");
            String tableName = eventArray[0];
            String eventType = eventArray[1];
            // 获取具体数据
            JSONObject valueStr = (JSONObject) result.get("value");
            // 转化为对应格式的json字符串
            JSONObject object = getESObject(valueStr, tableName);
            // 获取ES的type
            switch (eventType) {
                case "insert":
                case "update": {
                    esServiceImpl.addQuestionToEs(object.get("id").toString(), tableName, object);
                    break;
                }
                case "delete": {
                    // 删除videos
                    esServiceImpl.deleteQuestion(object.get("id").toString(), tableName);
                    break;
                }
            }
        }
    }
}
